package com.denghq.projectbuilder.component.memcache.configuration;

import lombok.extern.slf4j.Slf4j;
import net.spy.memcached.*;
import net.spy.memcached.spring.MemcachedClientFactoryBean;
import net.spy.memcached.transcoders.SerializingTranscoder;

import java.util.concurrent.ConcurrentHashMap;

@Slf4j
public class MemcachedUtils {


    private static final int DEFAULT_CAPACITY = 512;

    // 使用默认容量创建一个Map
    private static final ConcurrentHashMap<String, MemcachedClientFactoryBean> cache = new ConcurrentHashMap<>(
            DEFAULT_CAPACITY);

    public final static int DEFAULT_PORT = 11211;

    public final static String CACHE_KEY_PREFIX = "MemcachedUtils_";

    //private static int CACHE_EXPIRE_TIME = 1 * 24 * 60 * 60; //1天，client 缓存过期时间，单位秒
    //private static int CACHE_EXPIRE_TIME = 2; //1天，client 缓存过期时间，单位秒

    public final static ByteTranscoder byteTranscoder = new ByteTranscoder();

    public final static Object LOCK = new Object();

    /**
     * 获取
     *
     * @param ip
     * @param port
     * @param key
     * @return
     */
    public static byte[] getBytes(String ip, Integer port, String key) {
        String cacheKey = getCacheKey(ip, port);
        MemcachedClientFactoryBean client = null;
        try {
            client = cache.get(cacheKey);
            if (client == null) {
                synchronized (LOCK) {
                    client = cache.get(cacheKey);
                    if (client == null) {
                        client = createClient(ip, port == null ? DEFAULT_PORT : port);
                        //client = new MemcachedClient(new InetSocketAddress(ip, port == null ? DEFAULT_PORT : port));
                        cache.put(cacheKey, client);
                    }
                }
            }
            synchronized (client) {
                return (byte[]) ((MemcachedClient) client.getObject()).get(key, byteTranscoder);
                //return (byte[]) client.get(key, byteTranscoder);
            }
        } catch (Exception e) {
            log.error("连接Memcached失败，ip:{},port:{},msg:{}", ip, port, e.toString());
            //出现异常，从缓存清理掉对应的client,以便下次重新获取
            cache.put(cacheKey, null);
            if (client != null) {
                try {
                    //client.shutdown();
                    client.destroy();
                    client = null;
                } catch (Exception e1) {
                    log.error("关闭MemcachedClient出错，msg:{}", e1.getMessage());
                }

            }
            return null;
        }
    }


    /**
     * 获取缓存
     *
     * @param ip
     * @param port
     * @return
     */
    private static String getCacheKey(String ip, Integer port) {
        return CACHE_KEY_PREFIX + ip + "_" + (port == null ? DEFAULT_PORT : port);
    }

    private static MemcachedClientFactoryBean createClient(String ip, Integer port) throws Exception {
        MemcachedClientFactoryBean memcachedClientFactoryBean = new MemcachedClientFactoryBean();
        memcachedClientFactoryBean.setServers(ip + ":" + port);
        memcachedClientFactoryBean.setProtocol(ConnectionFactoryBuilder.Protocol.BINARY);
        memcachedClientFactoryBean.setTranscoder(new ByteTranscoder());
        memcachedClientFactoryBean.setOpTimeout(3000);
        memcachedClientFactoryBean.setTimeoutExceptionThreshold(3100);
        memcachedClientFactoryBean.setHashAlg(DefaultHashAlgorithm.KETAMA_HASH);
        memcachedClientFactoryBean.setLocatorType(ConnectionFactoryBuilder.Locator.CONSISTENT);
        memcachedClientFactoryBean.setFailureMode(FailureMode.Redistribute);
        memcachedClientFactoryBean.setUseNagleAlgorithm(false);
        memcachedClientFactoryBean.afterPropertiesSet();
        return memcachedClientFactoryBean;
    }
}


class ByteTranscoder extends SerializingTranscoder {
    @Override
    public Object decode(CachedData cachedData) {
        byte[] data = cachedData.getData();
        return data;
    }
}
